package com.atgugu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atgugu.realtime.app.BaseAppV2;
import com.atgugu.realtime.bean.OrderDetail;
import com.atgugu.realtime.bean.OrderInfo;
import com.atgugu.realtime.bean.OrderWide;
import com.atgugu.realtime.common.Constant;
import com.atgugu.realtime.util.DimUtil;
import com.atgugu.realtime.util.JdbcUtil;
import com.atgugu.realtime.util.RedisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.HashMap;

import static com.atgugu.realtime.common.Constant.TOPIC_DWD_ORDER_DETAIL;
import static com.atgugu.realtime.common.Constant.TOPIC_DWD_ORDER_INFO;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/12/28 14:19
 */
public class DwmOrderWideApp_Cache extends BaseAppV2 {
    public static void main(String[] args) {
        new DwmOrderWideApp_Cache().init(3003, 1, "DwmOrderWideApp", "DwmOrderWideApp",
                                         TOPIC_DWD_ORDER_INFO, TOPIC_DWD_ORDER_DETAIL);
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env,
                       HashMap<String, DataStreamSource<String>> topicToStream) {
        // 1. 事实表的join
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims = factsJoin(topicToStream);
        
        // 2. join维度数据
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithDims = factDims(orderWideStreamWithoutDims);
        orderWideStreamWithDims.print();
        
        // 3. 把宽表数据写入到Kafka中
    }
    
    private SingleOutputStreamOperator<OrderWide> factDims(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims) {
        /*
        每来一条数据, 都通过jdbc去Phoenix中查找对应的维度数据: 6张表
        
         */
        return orderWideStreamWithoutDims.map(new RichMapFunction<OrderWide, OrderWide>() {
            
            private Jedis jedis;
            private Connection phoenixConn;
            
            @Override
            public void open(Configuration parameters) throws Exception {
                phoenixConn = JdbcUtil.getJdbcConnection(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL);
                
                jedis = RedisUtil.getRedisClient();
            }
            
            @Override
            public void close() throws Exception {
                if (phoenixConn != null) {
                    phoenixConn.close();
                }
                
                if (jedis != null) {
                    jedis.close();  // 不一定是关闭客户端.
                    // 如果客户端是 new Jedis() 得到的, 则是关闭客户端
                    // 如果客户端是 通过连接池得到, 则是归还客户端
                }
                
            }
            
            @Override
            public OrderWide map(OrderWide orderWide) throws Exception {
                // 执行6个sql去查找对应的维度数据
                // 1. 补充userInfo
                JSONObject userInfo = DimUtil.readDim(phoenixConn, jedis, "dim_user_info", orderWide.getUser_id());
                orderWide.setUser_gender(userInfo.getString("GENDER"));
                orderWide.calcUserAge(userInfo.getString("BIRTHDAY"));
                
                // 2. 补充省份
                JSONObject provinceInfo = DimUtil.readDim(phoenixConn, jedis, "dim_base_province", orderWide.getProvince_id());
                orderWide.setProvince_3166_2_code(provinceInfo.getString("ISO_3166_2"));
                orderWide.setProvince_area_code(provinceInfo.getString("AREA_CODE"));
                orderWide.setProvince_iso_code(provinceInfo.getString("ISO_CODE"));
                orderWide.setProvince_name(provinceInfo.getString("NAME"));
                
                // 3. 补充sku
                JSONObject skuInfo = DimUtil.readDim(phoenixConn, jedis, "dim_sku_info", orderWide.getSku_id());
                orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
                orderWide.setOrder_price(skuInfo.getBigDecimal("PRICE"));
                
                orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                orderWide.setTm_id(skuInfo.getLong("TM_ID"));
                orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                
                // 4. 补充spu
                JSONObject spuInfo = DimUtil.readDim(phoenixConn, jedis, "dim_spu_info", orderWide.getSpu_id());
                orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));
                // 5. 补充tm
                JSONObject tmInfo = DimUtil.readDim(phoenixConn, jedis, "dim_base_trademark", orderWide.getTm_id());
                orderWide.setTm_name(tmInfo.getString("TM_NAME"));
                // 5. 补充c3
                JSONObject c3Info = DimUtil.readDim(phoenixConn, jedis, "dim_base_category3", orderWide.getCategory3_id());
                orderWide.setCategory3_name(c3Info.getString("NAME"));
                
                return orderWide;
            }
        });
    }
    
    private SingleOutputStreamOperator<OrderWide> factsJoin(HashMap<String, DataStreamSource<String>> topicToStream) {
        // interval join : 1. 只支持事件时间  2. 必须keyBy之后使用
        
        KeyedStream<OrderInfo, Long> orderInfoStream = topicToStream
            .get(TOPIC_DWD_ORDER_INFO)
            .map(info -> JSON.parseObject(info, OrderInfo.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((info, ts) -> info.getCreate_ts())
            
            )
            .keyBy(OrderInfo::getId);
        
        KeyedStream<OrderDetail, Long> orderDetailStream = topicToStream
            .get(TOPIC_DWD_ORDER_DETAIL)
            .map(info -> JSON.parseObject(info, OrderDetail.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((detail, ts) -> detail.getCreate_ts())
            
            )
            .keyBy(OrderDetail::getOrder_id);
        
        return orderInfoStream
            .intervalJoin(orderDetailStream)
            .between(Time.seconds(-5), Time.seconds(5))
            .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                @Override
                public void processElement(OrderInfo left,
                                           OrderDetail right,
                                           Context ctx,
                                           Collector<OrderWide> out) throws Exception {
                    out.collect(new OrderWide(left, right));
                }
            });
        
    }
}
/*
缓存

1. 使用 flink 的状态?
优点
    1. flink的内部, 状态的存取方便, 全部有flink管理
    2. 如果使用rocksdb, 存储量可以很大
    3. 不用走网络

缺点
    1. 会占用一部分flink的内存, 导致flink计算的内存减少
    2. 状态中的维度数据没有办法翻译维度数据的变更

2. 使用redis?
优点
    1. 专门的服务器部署redis, 内存可以很大
    2. 缓存中的维度数据能及时变化
缺点
    读写数据需要通过网络


选择redis
旁路缓存
 */